package com.swak.reactivex.transport;

import static com.swak.Constants.Default_Timeout;
import static com.swak.Constants.Timeout;

import java.io.File;
import java.io.FileNotFoundException;
import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.registry.URL;

public class HeaderExchangeChannel implements ExchangeChannel {

	private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeChannel.class);

	private static final String CHANNEL_KEY = HeaderExchangeChannel.class.getName() + ".CHANNEL";

	private final Channel channel;

	HeaderExchangeChannel(Channel channel) {
		if (channel == null) {
			throw new IllegalArgumentException("channel == null");
		}
		this.channel = channel;
	}

	@Override
	public URL url() {
		return channel.url();
	}

	@Override
	public boolean isConnected() {
		return channel.isConnected();
	}

	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		if (!this.isConnected()) {
			throw new RemotingException(
					"Failed to send message " + message + ", cause: The channel " + this + " is closed!");
		}

//		/*
//		 * 一定要封装为 Message 类型
//	     * 这里不能限制的这么死，需要支持多种数据类型
//		 */
//		Message request = (message instanceof Message) ? // 判断是否为 Message 类型
//				(Message) message : //
//				Message.forMessage().setData(message); // 简单的Message 类型

		/*
		 * 发送请求
		 */
		channel.send(message, sent);
	}

	@Override
	public void sendFile(File file, boolean sent) throws RemotingException, FileNotFoundException {
		if (!this.isConnected()) {
			throw new RemotingException("Failed to send file " + file + ", cause: The channel " + this + " is closed!");
		}
		channel.sendFile(file, sent);
	}

	@Override
	public CompletableFuture<Object> request(Object request) throws RemotingException {
		return request(request, channel.url().getIntParameter(Timeout, Default_Timeout));
	}

	@Override
	public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
		if (!this.isConnected()) {
			throw new RemotingException(
					"Failed to send request " + request + ", cause: The channel " + this + " is closed!");
		}
		// create request.
		Message req = (request instanceof Message) ? (Message) request : Message.forRequest().setData(request);
		if (!req.isRequest() || !req.isTwoWay()) {
			throw new RemotingException(
					"Cannot send message, cause: The Message " + request + " is not a twoway request!");
		}
		TimeoutFuture future = TimeoutFuture.newFuture(channel, req, timeout);
		try {
			channel.send(req);
		} catch (RemotingException e) {
			future.cancel();
			throw e;
		}
		return future;
	}

	@Override
	public void close() {
		try {
			// graceful close
			TimeoutFuture.closeChannel(channel);
			channel.close();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	@Override
	public InetSocketAddress localAddress() {
		return channel.localAddress();
	}

	@Override
	public InetSocketAddress remoteAddress() {
		return channel.remoteAddress();
	}

	@Override
	public boolean hasAttribute(String key) {
		return channel.hasAttribute(key);
	}

	@Override
	public Object getAttribute(String key) {
		return channel.getAttribute(key);
	}

	@Override
	public void setAttribute(String key, Object value) {
		channel.setAttribute(key, value);
	}

	@Override
	public void removeAttribute(String key) {
		channel.removeAttribute(key);
	}

	static HeaderExchangeChannel getOrAddChannel(Channel ch) {
		if (ch == null) {
			return null;
		}
		HeaderExchangeChannel ret = (HeaderExchangeChannel) ch.getAttribute(CHANNEL_KEY);
		if (ret == null) {
			ret = new HeaderExchangeChannel(ch);
			if (ch.isConnected()) {
				ch.setAttribute(CHANNEL_KEY, ret);
			}
		}
		return ret;
	}

	static void removeChannelIfDisconnected(Channel ch) {
		if (ch != null && !ch.isConnected()) {
			ch.removeAttribute(CHANNEL_KEY);
		}
	}

	static void removeChannel(Channel ch) {
		if (ch != null) {
			ch.removeAttribute(CHANNEL_KEY);
		}
	}

}
